package cn.doitedu.udfs;

import cn.doitedu.pojo.*;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class MatchFunctionV2 extends KeyedProcessFunction<String, LogBean, MatchResult> {

    private Connection hbaseConn;
    private java.sql.Connection clickHouseConn;
    private RuleCondition ruleCondition;

    @Override
    public void open(Configuration parameters) throws Exception {
        //在TaskManager初始化subtask时，获取全局参数
        ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();

        //初始化连接
        //1.创建Hbase的连接
        org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
        hbaseConf.set("hbase.zookeeper.quorum", parameterTool.getRequired("hbase.zookeeper.quorum"));
        hbaseConf.setInt("hbase.zookeeper.property.clientPort", parameterTool.getInt("hbase.zookeeper.property.clientPort", 2181));
        hbaseConn = ConnectionFactory.createConnection(hbaseConf);

        clickHouseConn = DriverManager.getConnection(parameterTool.getRequired("clickhouse.url"));

        //创建一个RuleCondition，以后使用广播状态动态广播获取
        ruleCondition = new RuleCondition();
        //模拟一个规则，以后是通过广播状态动态接入
        //运营人员，事先定义好的规则
        //触发事件参数
        EventParam eventParam = new EventParam();
        eventParam.setEventId("G");
        HashMap<String, String> props = new HashMap<>();
        props.put("p4", "v1");
        props.put("p9", "v1");
        eventParam.setProps(props);
        ruleCondition.setTriggerEvent(eventParam);
        //用户画像参数
        HashMap<String, String> userProfiles = new HashMap<>();
        userProfiles.put("tag1", "v2");
        userProfiles.put("tag3", "v2");
        ruleCondition.setUserProfiles(userProfiles);
        //历史行为次数参数
        EventCountCondition eventCountCondition = new EventCountCondition();
        eventCountCondition.setEventId("G");
        HashMap<String, String> eventCountProps = new HashMap<>();
        eventCountProps.put("p4", "v1");
        eventCountCondition.setProps(eventCountProps);
        eventCountCondition.setStartTime(1649433600000L);
        eventCountCondition.setEndTime(1649606399000L);
        eventCountCondition.setCount(2);
        ruleCondition.setEventCountCondition(eventCountCondition);
        //设置行为序列参数
        EventSeqCondition eventSeqCondition = new EventSeqCondition();
        List<String> events = Arrays.asList("A", "H", "A");
        eventSeqCondition.setEvents(events);
        eventSeqCondition.setStartTime(1649433600000L);
        eventSeqCondition.setEndTime(1649606399000L);
        ruleCondition.setEventSeqCondition(eventSeqCondition);
    }

    @Override
    public void processElement(LogBean bean, Context ctx, Collector<MatchResult> out) throws Exception {


        //1.匹配触发事件
        boolean matchEvent = isMatchEvent(bean, ruleCondition.getTriggerEvent()); //行为ID和行为属性
        if (!matchEvent) {
            return;
        }
        //2.匹配用户画像
        boolean matchProfile = isMatchProfile(bean, ruleCondition.getUserProfiles()); //用户画像数据
        if (!matchProfile) {
            return;
        }
        //3.匹配行为次数
        boolean matchEventCount = isMatchEventCount(bean, ruleCondition.getEventCountCondition()); //历史行为次数
        if (!matchEventCount) {
            return;
        }

        //4.匹配行为序列
        boolean matchEventSeq = isMatchEventSeq(bean, ruleCondition.getEventSeqCondition()); //行为序列（两个或两个以上的行为）
        if (matchEventSeq) {
            out.collect(new MatchResult(bean.getDeviceId(), bean.getTimeStamp(), "rule001", System.currentTimeMillis()));
        }

    }

    private boolean isMatchEvent(LogBean bean, EventParam eventParam) {
        //完成匹配触发事件的逻辑
        //eventParam中封装着事先定义好的触发事件ID和属性
        String eventId = eventParam.getEventId();
        Map<String, String> props = eventParam.getProps();
        //用户的实际的事件ID
        //用户实际的行为属性
        Map<String, String> properties = bean.getProperties();
        if (eventId.equals(bean.getEventId())) {
            //匹配行为属性
            for (Map.Entry<String, String> entry : props.entrySet()) {
                //事先定义好的行为属性名称和值
                String targetKey = entry.getKey();
                String targetVal = entry.getValue();
                //根据事先定义好的属性key，取出实际key的value
                String val = properties.get(targetKey);
                if (!targetVal.equals(val)) {
                    return false;
                }
            }
        } else {
            return false;
        }
        return true;
    }

    //是否匹配画像
    private boolean isMatchProfile(LogBean bean, Map<String, String> userProfiles) throws Exception {

        String deviceId = bean.getDeviceId();
        Table table = hbaseConn.getTable(TableName.valueOf("eagle_profile"));
        try {
            Get get = new Get(Bytes.toBytes(deviceId));
            //添加过滤条件
            for (String key : userProfiles.keySet()) {
                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes(key));
            }
            Result result = table.get(get);
            if (result != null) {
                for (String tag : userProfiles.keySet()) {
                    String val = new String(result.getValue(Bytes.toBytes("f"), Bytes.toBytes(tag)));
                    //取出事先定义好的该tag的值
                    String targetVal = userProfiles.get(tag);
                    if (!targetVal.equals(val)) {
                        return false;
                    }
                }
            } else {
                return false;
            }
        } finally {
            table.close();
        }
        return true;
    }

    //查询ClickHouse匹配历史行为次数
    private boolean isMatchEventCount(LogBean bean, EventCountCondition eventCountCondition) throws Exception {
        String deviceId = bean.getDeviceId();

        String sql = "select count(*) from eagle_detail where deviceId = ? and timeStamp >= ? and timeStamp < ? and eventId = ?";
        Map<String, String> props = eventCountCondition.getProps();
        for (Map.Entry<String, String> entry : props.entrySet()) {
            String key = entry.getKey();
            String value = entry.getValue();
            sql += " and properties['" + key + "'] = '" + value + "'";
        }
        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        try {
            preparedStatement = clickHouseConn.prepareStatement(sql);
            preparedStatement.setString(1, deviceId);
            preparedStatement.setLong(2, eventCountCondition.getStartTime());
            preparedStatement.setLong(3, eventCountCondition.getEndTime());
            preparedStatement.setString(4, eventCountCondition.getEventId());

            resultSet = preparedStatement.executeQuery();
            long count = 0;
            if (resultSet.next()) {
                count = resultSet.getLong(1);
            }
            if (count < eventCountCondition.getCount()) {
                return false;
            }
        } finally {
            if (resultSet != null) {
                resultSet.close();
            }

            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
        return true;
    }

    private boolean isMatchEventSeq(LogBean bean, EventSeqCondition eventSeqCondition) throws Exception {
        StringBuffer sb = new StringBuffer();
        sb.append("SELECT sequenceMatch('");
        List<String> events = eventSeqCondition.getEvents();
        for (int i = 0; i < events.size(); i++) {
            sb.append("(?" + (i+1) +")");
            if (i == events.size() - 1) {
                sb.append("')");
            }
        }
        sb.append("(toDateTime(`timeStamp`)");
        for (int i = 0; i < events.size(); i++) {
            sb.append(", eventId = '" + events.get(i) + "'");
            if (i == events.size() - 1) {
                sb.append(")");
            }
        }
        sb.append(" AS is_match FROM eagle_detail where deviceId = ? and timeStamp >= ? and timeStamp < ? GROUP BY deviceId");

        PreparedStatement preparedStatement = null;
        ResultSet resultSet = null;
        try {
            preparedStatement = clickHouseConn.prepareStatement(sb.toString());
            preparedStatement.setString(1, bean.getDeviceId());
            preparedStatement.setLong(2, eventSeqCondition.getStartTime());
            preparedStatement.setLong(3, eventSeqCondition.getEndTime());


            resultSet = preparedStatement.executeQuery();
            long count = 0;
            if (resultSet.next()) {
                count = resultSet.getLong(1);
            }
            if (count == 0) {
                return false;
            }
        } finally {
            if (resultSet != null) {
                resultSet.close();
            }

            if (preparedStatement != null) {
                preparedStatement.close();
            }
        }
        return true;
    }


    @Override
    public void close() throws Exception {
        //释放资源
        hbaseConn.close();
        clickHouseConn.close();

    }
}
